package spark.core.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 常用的Transformation算子
 */
public class TransformationOperation {

    public static void main(String[] args) {
//        map();
//        filter();
        cogroup();
    }


    private static  void map(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("map");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> numbers = Arrays.asList(1,2,3,4,5);
        //并行化集合，创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
        JavaRDD<Integer> newNumberRDD = numberRDD.map(new Function<Integer, Integer>() {
            public Integer call(Integer v1) throws Exception {
                return v1 * 2;
            }
        });
        newNumberRDD.foreach(new VoidFunction<Integer>() {
            public void call(Integer v1) throws Exception {
                System.out.println("use map funtion,the list element is "+ v1);
            }
        });
        sc.close();
    }

    private static  void flatMap(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("flatMap");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> lines = Arrays.asList("hello me", "hello you", "hello world");
        //并行化集合，创建初始RDD
        JavaRDD<String> words = sc.parallelize(lines);
        //flatMap算子，在JAVA中，接收的参数是flatmapfunction
        //我们需要自己定义FlatmapFunction的第二个泛型类型，即，代表了返回的新元素的类型
        //flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，返回可以返回多个元素，
        //多个元素，即封装在Iterable集合中，可以使用ArrayList等集合
        //新的RDD中，即封装在所有的新元素。也就是说，新的RDD的太小一定>=原始RDD
        JavaRDD<String> flatMapRDD =  words.flatMap(new FlatMapFunction<String, String>() {
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" "));
            }
        });
        flatMapRDD.foreach(new VoidFunction<String>() {
            public void call(String v1) throws Exception {
                System.out.println("use flatMap funtion,the list element is " + v1);
            }
        });
        sc.close();
    }


    private static  void filter(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("filter");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
        //并行化集合，创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
        JavaRDD<Integer> newNumberRDD = numberRDD.filter(new Function<Integer, Boolean>() {
            public Boolean call(Integer v1) throws Exception {
                return v1 % 2 == 0;
            }
        });
        newNumberRDD.foreach(new VoidFunction<Integer>() {
            public void call(Integer v1) throws Exception {
                System.out.println("use filter funtion,the list element is "+ v1);
            }
        });
        sc.close();
    }


    /**
     * 按照班級對成績進行分組
     */
    private static  void groupByKey(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("groupByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<String,Integer>> numbers = Arrays.asList(
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class1", 75),
                new Tuple2<String, Integer>("class2", 65),
                new Tuple2<String, Integer>("class1", 77));
        //并行化集合，创建初始RDD
        JavaPairRDD<String,Integer> scores = sc.parallelizePairs(numbers);
        JavaPairRDD<String,Iterable<Integer>> result =scores.groupByKey();
        result.foreach(new VoidFunction< Tuple2<String, Iterable<Integer>>>() {
            public void call(Tuple2<String, Iterable<Integer>> t) throws Exception {
                System.out.println("use groupByKey funtion,the list element is "+ t._1()+":"+t._2());
                Iterator<Integer> ite = t._2().iterator();
                while(ite.hasNext()){
                    System.out.println("scores is "+ ite.next());
                }
            }
        });
        sc.close();
    }



    /**
     * 统计每个班级的总分
     */
    private static  void reduceByKey(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("reduceByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<String,Integer>> numbers = Arrays.asList(
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class1", 75),
                new Tuple2<String, Integer>("class2", 65),
                new Tuple2<String, Integer>("class1", 77));
        //并行化集合，创建初始RDD
        JavaPairRDD<String,Integer> scores = sc.parallelizePairs(numbers);
        JavaPairRDD<String,Integer> result =scores.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        result.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            public void call(Tuple2<String, Integer> v) throws Exception {
                System.out.println("the Class is "+ v._1()+" total scores is "+ v._2());
            }
        });
        sc.close();
    }



    /**
     * 统计每个班级的总分
     */
    private static  void sortByKey(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("sortByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> numbers = Arrays.asList(
                new Tuple2<Integer, String>(75, "class1"),
                new Tuple2<Integer, String>(78,  "class1"),
                new Tuple2<Integer, String>(56, "class2"),
                new Tuple2<Integer, String>(89, "class2"));
        //并行化集合，创建初始RDD
        JavaPairRDD<Integer, String> scores = sc.parallelizePairs(numbers);
        JavaPairRDD<Integer, String> result =scores.sortByKey(false);
        result.foreach(new VoidFunction<Tuple2<Integer, String>>() {
            public void call(Tuple2<Integer, String> v) throws Exception {
                System.out.println("the Class is "+ v._2()+"  scores is "+ v._1());
            }
        });
        sc.close();
    }


    /**
     * join 打印学生成绩
     */
    private static  void join(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("sortByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> list1 = Arrays.asList(
                new Tuple2<Integer, String>(1, "Eric"),
                new Tuple2<Integer, String>(2,  "Tom"),
                new Tuple2<Integer, String>(3, "laurance"),
                new Tuple2<Integer, String>(4, "mike"));

        List<Tuple2<Integer, Integer>> list2 = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 45),
                new Tuple2<Integer, Integer>(2, 78),
                new Tuple2<Integer, Integer>(3, 68),
                new Tuple2<Integer, Integer>(4, 97) ,
                new Tuple2<Integer, Integer>(1, 45),
                new Tuple2<Integer, Integer>(2, 78),
                new Tuple2<Integer, Integer>(3, 68),
                new Tuple2<Integer, Integer>(4, 97));
        //并行化集合，创建初始RDD
        JavaPairRDD<Integer, String> rdd1 = sc.parallelizePairs(list1);
        JavaPairRDD<Integer, Integer> rdd2 = sc.parallelizePairs(list2);

       JavaPairRDD<Integer,Tuple2<String, Integer>>  studentScores =  rdd1.join(rdd2);
        studentScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, Integer>>>() {
            public void call(Tuple2<Integer, Tuple2<String, Integer>> s) throws Exception {
                System.out.println("NO."+s._1()+",name:"+s._2()._1()+" score :"+ s._2()._2());
            }
        });
        sc.close();
    }


    /**
     * join 打印学生成绩
     */
    private static  void cogroup(){
        SparkConf conf = new SparkConf().setMaster("local").setAppName("sortByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> list1 = Arrays.asList(
                new Tuple2<Integer, String>(1, "Eric"),
                new Tuple2<Integer, String>(2,  "Tom"),
                new Tuple2<Integer, String>(3, "laurance"),
                new Tuple2<Integer, String>(4, "mike"));

        List<Tuple2<Integer, Integer>> list2 = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 45),
                new Tuple2<Integer, Integer>(2, 78),
                new Tuple2<Integer, Integer>(3, 68),
                new Tuple2<Integer, Integer>(4, 97) ,
                new Tuple2<Integer, Integer>(1, 45),
                new Tuple2<Integer, Integer>(2, 78),
                new Tuple2<Integer, Integer>(3, 68),
                new Tuple2<Integer, Integer>(4, 97));
        //并行化集合，创建初始RDD
        JavaPairRDD<Integer, String> rdd1 = sc.parallelizePairs(list1);
        JavaPairRDD<Integer, Integer> rdd2 = sc.parallelizePairs(list2);

        JavaPairRDD<Integer,Tuple2<Iterable<String>, Iterable<Integer>>>  studentScores =  rdd1.cogroup(rdd2);
        studentScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>>>() {
         public void call(Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> v) throws Exception {
             System.out.println("id:"+ v._1());
             System.out.println("name:"+ v._2()._1());
             System.out.println("score:"+ v._2()._2());

                                  }
                              });
                sc.close();
    }

}
